package com.lhf.azkaban.springbatch.example.batchconfig;

import com.lhf.azkaban.springbatch.example.batchprocessing.RepayOrderProcessor;
import com.lhf.azkaban.springbatch.example.batchreader.RepayOrderFlatFileItemReader;
import com.lhf.azkaban.springbatch.example.batchwriter.RepayOrderMybatisWrite;
import com.lhf.azkaban.springbatch.example.job.config.BaseConfig;
import com.lhf.azkaban.springbatch.example.mode.RepayOrderInfo;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.JobRegistry;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.explore.JobExplorer;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
import org.springframework.batch.support.DatabaseType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;

/**
 * @Author:95780
 * @Date: 11:17 2019/11/26
 * @Description:
 */
@Configuration
@EnableBatchProcessing
@EnableAutoConfiguration
public class StartJobOperatorConfig extends BaseConfig {
    @Autowired
    private JobRepository jobRepository;

    @Autowired
    private JobExplorer jobExplorer;

    @Autowired
    private JobRegistry jobRegistry;


    @Autowired
    public DataSource dataSource;

    @Autowired
    private RepayOrderFlatFileItemReader repayOrderFlatFileItemReader;

    @Autowired
    private RepayOrderMybatisWrite repayOrderMybatisWrite;

    //创建任务对象
    @Bean
    public Job repayOrderJob() {
        return jobBuilderFactory.get("repayOrderJob")
                .incrementer(new RunIdIncrementer())
                .listener(myJobListener) /*增加对job的监听*/
                .flow(repayOrderStep())
                .end()
                .build();
    }

    @Bean
    public Step repayOrderStep() {
        return stepBuilderFactory.get("repayOrderStep")
                .<RepayOrderInfo, RepayOrderInfo>chunk(100) //读完几条数据后进行数据的处理；
                .reader(repayOrderFlatFileItemReader) //读数据，实现reader接口的对象类型；
                .processor(repayOrderProcessor())
                .writer(repayOrderMybatisWrite)
                .build();
    }


    @Bean
    JobRepository obRepository(PlatformTransactionManager platformTransactionManager) throws Exception {
        JobRepositoryFactoryBean jobRepositoryFactoryBean = new JobRepositoryFactoryBean();
        jobRepositoryFactoryBean.setDatabaseType(DatabaseType.MYSQL.getProductName());
        //配置指定的dataSource
        jobRepositoryFactoryBean.setDataSource(dataSource);
        jobRepositoryFactoryBean.setTransactionManager(platformTransactionManager);
        return jobRepositoryFactoryBean.getObject();
    }

//    @Bean
//    public JobOperator jobOperator() {
//        SimpleJobOperator operator = new SimpleJobOperator();
//        operator.setJobLauncher(jobLauncher);
//        operator.setJobParametersConverter(new DefaultJobParametersConverter());
//        operator.setJobRepository(jobRepository);
//        operator.setJobExplorer(jobExplorer);
//        operator.setJobRegistry(jobRegistry);
//        return (JobOperator) operator;
//    }



    //任务的执行由step决定，实现数据的读取、处理和输出等；


    @Bean
    public RepayOrderProcessor repayOrderProcessor() {
        return new RepayOrderProcessor();
    }

}
